/**********************************最简单的Demo****************************************/
/**
 * 服务端
 */
public class NettyServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    Channel channel = ctx.channel();
                                    ByteBuf buf = (ByteBuf) msg;
                                    System.out.println("客户端的地址：" + channel.remoteAddress());
                                    System.out.println("服务端收到的消息：" + buf.toString(CharsetUtil.UTF_8));
                                }
                            });
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

/**
 * 客户端
 */

public class NettyClient {
    public static void main(String[] args) {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        try {
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    Channel channel = ctx.channel();
                                    channel.writeAndFlush(Unpooled.copiedBuffer("Hello,server!", CharsetUtil.UTF_8));
                                }
                            });
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

/**********************************Netty中NioEventLoopGroup创建过程分析****************************************/

NioEventLoopGroup workerGroup = new NioEventLoopGroup();
// 这个构造器的调用链比较长，主要在MultithreadEventLoopGroup类中，这里省略了部分中间调用
public NioEventLoopGroup() {
    this(0);
}

protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
    super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}

// 可以看见默认的线程数是DEFAULT_EVENT_LOOP_THREADS，根据DEFAULT_EVENT_LOOP_THREADS的初始定义，其等于CPU核心数*2
private static final int DEFAULT_EVENT_LOOP_THREADS;

static {
    DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
            "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

    if (logger.isDebugEnabled()) {
        logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
    }
}

// 最后构造器的调用链会进入MultithreadEventExecutorGroup的构造方法
// 在该构造器中，首先会创建一个executor，这个executor会传递给创建的每一个NioEventLoop，其作用是进行任务的异步提交

if (executor == null) {
    executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}

// 创建好executor后会进行循环的创建NioEventLoop，其中调用的NioEventLoopGroup#newChild方法
children[i] = newChild(executor, args);

protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
        ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
}
// 在NioEventLoopGroup#newChild方法中实质是调用的是NioEventLoop的构造方法
// 在NioEventLoop的构造方法中首先会调用超类的构造，并调用NioEventLoop#newTaskQueue构造好任务队列
// NioEventLoop构造方法节选
super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
                rejectedExecutionHandler);
// NioEventLoop#newTaskQueue方法默认会构建容量为Integer.MAX_VALUE的MPSC队列
// NioEventLoop#newTaskQueue方法
private static Queue<Runnable> newTaskQueue(
    EventLoopTaskQueueFactory queueFactory) {
    if (queueFactory == null) {
        return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);
    }
    return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
}
// NioEventLoop#newTaskQueue0方法
private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {
    // This event loop never calls takeTask()
    return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()
            : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);
}                

// 接着在构造方法中会创建两个selector，其中需要关注NioEventLoop#openSelector方法
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
unwrappedSelector = selectorTuple.unwrappedSelector;
// 在NioEventLoop#openSelector方法中首先会创建JDK中的selector
unwrappedSelector = provider.openSelector();
// 然后创建一个Netty中的SelectedSelectionKeySet并使用暴力反射的方式将JDK中SelectionKey底层的Set替换成数组
// 这样做的目的是也是出于数组遍历更快的考量（Set底层是哈希表）
// 这也就说明了为什么在创建NioEventLoop时会创建两个selector
final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet();
Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);
if (cause != null) {
    return cause;
}
cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);
if (cause != null) {
    return cause;
}

selectedKeysField.set(unwrappedSelector, selectedKeySet);
publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);

// 在创建完NioEventLoop后，会创建一个chooser选择器
// MultithreadEventExecutorGroup构造方法节选
chooser = chooserFactory.newChooser(children);
// 创建的选择器只有一个next方法，其作用是根据executor来选择执行的NioEventLoop
// PowerOfTwoEventExecutorChooser#next方法
public EventExecutor next() {
    return executors[idx.getAndIncrement() & executors.length - 1];
}

/**********************************Netty服务端bind源码分析****************************************/
ChannelFuture channelFuture = serverBootstrap.bind(8888).sync();

// bind方法最终会进入AbstractBootstrap#doBind方法
// 在doBind方法中主要做三件事：1、创建并初始化NioSocketChannel；2、注册NioSocketChannel；3、进行端口绑定
// 其中1和2体现在AbstractBootstrap#initAndRegister方法中，下面将做详细分析
// AbstractBootstrap#doBind方法节选
final ChannelFuture regFuture = initAndRegister();

/**********************************newChannel源码分析****************************************/
// 在AbstractBootstrap#initAndRegister方法中首先是创建并初始化channel，其中channel的创建是使用channelFactory进行创建，channelFactory的创建是在配置启动类的channel的时候进行创建的
// AbstractBootstrap#initAndRegister方法节选 
channel = channelFactory.newChannel();
// AbstractBootstrap#channel方法
public B channel(Class<? extends C> channelClass) {
    return channelFactory(new ReflectiveChannelFactory<C>(
            ObjectUtil.checkNotNull(channelClass, "channelClass")
    ));
}
// 在ReflectiveChannelFactory#newChannel方法中最终使用反射调用构造器来进行channel的创建，因为这里传入的channel类型是NioServerSocketChannel，所以直接定位到NioServerSocketChannel的构造方法
// ReflectiveChannelFactory#newChannel节选
constructor.newInstance();

// NioServerSocketChannel的构造方法进行重载，首先会调用NioServerSocketChannel#newSocket方法创建一个JDK中的ServerSocketChannel
public NioServerSocketChannel() {
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
// NioServerSocketChannel#newSocket节选
provider.openServerSocketChannel()
// NioServerSocketChannel构造方法的重载，在该构造方法中会先调用超类的构造方法，最终会进入AbstractNioChannel的构造方法
public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
// 在AbstractNioChannel的构造方法中同样会进行父类构造的调用，同时将channel设置为非阻塞模式
// AbstractNioChannel构造方法节选
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
// 设置通道为非阻塞
ch.configureBlocking(false);
// AbstractNioChannel的父类构造会进入AbstractChannel的构造，在该构造函数中会为channel分配一个id，同时创建一个unsafe类和pipeline
// 其中unsafe类是底层的操作类，在连接的接入和消息的读取都会发挥作用，这个之后会分析；而pipeline就是Netty中事件传播管道
protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}
// 从DefaultChannelPipeline的构造方法可以看出构造好的pipeline默认就有两个元素：head和tail，pipeline可以看作一个双向链表，其中的元素可以看作是ChannelHandlerContext
// Netty事件的传播机制总是从pipeline的一端往另一端传播的
// DefaultChannelPipeline构造函数节选
tail = new TailContext(this);
head = new HeadContext(this);

// 到这里NioServerSocketChannel构造方法中的父类构造基本就已经走完，下面还有一个channel配置类的创建
// NioServerSocketChannel构造方法节选
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
// 对于NioServerSocketChannelConfig的构造，其中一个参数是channel本身，另一个参数是JDK中的ServerSocket。在配置类中创建了一个可重用的ByteBuf的分配器，可以进行直接内存的分配

// DefaultChannelConfig构造方法
public DefaultChannelConfig(Channel channel) {
    this(channel, new AdaptiveRecvByteBufAllocator());
}
protected DefaultChannelConfig(Channel channel, RecvByteBufAllocator allocator) {
    setRecvByteBufAllocator(allocator, channel.metadata());
    this.channel = channel;
}

// 这里需要明确一个概念，JDK中NIO是面向缓冲区操作，而这里的channel配置类其中一个功能是对channel缓冲区的配置，在Netty中使用直接内存作为channel的缓冲区，
// 这样做的原因也是因为直接内存操作减少了数据往堆内存的拷贝，同时减轻了GC的压力

/**********************************init源码分析****************************************/
// AbstractBootstrap#initAndRegister方法节选
init(channel);
// 相比于channel的创建，ServerBootstrap#init方法方法要简单许多。在ServerBootstrap#init方法中首先进来会根据创建启动类的配置进行channel的属性配置
// 比如配置服务端可连接的队列ChannelOption.SO_BACKLOG，是否可以重用地址ChannelOption.SO_REUSEADDR等
// ServerBootstrap#init方法节选
setChannelOptions(channel, options0().entrySet().toArray(newOptionArray(0)), logger);
setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0)));
// 在对channel的属性进行初始化后，另一件重要的事就是向pipeline中添加一个ChannelInitializer实现类，该ChannelInitializer实现类的调用是异步调用（并不在当前线程中调用，而是由Netty的NIO线程调用）
// 在ChannelInitializer中会添加一个接入器Acceptor，该接入器的作用是检测新连接的接入并创建一个客户端channel分配给workerGroup处理，后面会详细探讨

ChannelPipeline p = channel.pipeline();
p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});
// 至此，channel的初始化工作就算基本完成，其初始化工作主要分为以下几步
// 1. 对channel进行属性设置
// 2. 向关联的pipeline中添加一个ChannelInitializer，在ChannelInitializer中异步的加入一个接入器Acceptor负责新连接的接入处理

/**********************************register源码分析****************************************/
// AbstractBootstrap#initAndRegister方法节选
ChannelFuture regFuture = config().group().register(channel);
// 注意这里的config是启动类的配置，不是之前创建channel的config，config().group()返回是配置类的bossGroup，最终的register方法会进入MultithreadEventLoopGroup#register方法
// MultithreadEventLoopGroup#register方法，这里的next相当于调用chooser的next选择一个NioEventLoop进行注册操作
// MultithreadEventLoopGroup#register方法
@Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}
// 最终NioEventLoop的注册函数会调用SingleThreadEventLoop#register的注册方法
// SingleThreadEventLoop#register方法，在该方法中将channel和NioEventLoop包装成DefaultChannelPromise（Promise是Netty异步调用的关键，可以类比JDK中的Future）
// SingleThreadEventLoop#register方法
@Override
public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}
// SingleThreadEventLoop#register方法重载
@Override
public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    promise.channel().unsafe().register(this, promise);
    return promise;
}
// 在SingleThreadEventLoop#register方法中会调用AbstractChannel#register，在AbstractChannel#register首先会判断该channel是否注册过
// 然后调用SingleThreadEventLoop#inEventLoop方法判断当前执行线程是否是Netty的异步线程（当前事件是否是由绑定的NIO线程执行，如果不是则加入EventLoop的异步任务队列，
// SingleThreadEventLoop#inEventLoop方法是Netty无锁串行化实现的关键）
// 对于第一次的注册调用，都会进入else分支，将注册事件以任务的形式提交到EventLoop中，在任务中执行AbstractChannel#register0方法
// AbstractChannel#register方法节选
AbstractChannel.this.eventLoop = eventLoop;

if (eventLoop.inEventLoop()) {
    register0(promise);
} else {
    try {
        eventLoop.execute(new Runnable() {
            @Override
            public void run() {
                register0(promise);
            }
        });
    } catch (Throwable t) {
        logger.warn(
                "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                AbstractChannel.this, t);
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}
// 在else分支中会调用SingleThreadEventLoop#execute方法
// SingleThreadEventLoop#execute方法首先也会进行SingleThreadEventLoop#inEventLoop方法判断，然后将任务添加到任务队列中，然后调用SingleThreadEventExecutor#startThread方法启动线程
// SingleThreadEventLoop#execute方法
public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }

    boolean inEventLoop = inEventLoop();
    // 添加任务到任务队列中
    addTask(task);
    if (!inEventLoop) {
        // 启动线程，使用CAS操作保证只有一个线程操作成功，且避免重复启动
        startThread();
        if (isShutdown()) {
            boolean reject = false;
            try {
                if (removeTask(task)) {
                    reject = true;
                }
            } catch (UnsupportedOperationException e) {
                // The task queue does not support removal so the best thing we can do is to just move on and
                // hope we will be able to pick-up the task before its completely terminated.
                // In worst case we will log on termination.
            }
            if (reject) {
                reject();
            }
        }
    }

    if (!addTaskWakesUp && wakesUpForTask(task)) {
        wakeup(inEventLoop);
    }
}


// SingleThreadEventExecutor#startThread方法节选
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
    boolean success = false;
    try {
        doStartThread();
        success = true;
    }
}
// 在此之中需要重点关注SingleThreadEventExecutor#doStartThread方法，在该方法中会通过执行其创建一个线程并对thread字段进行赋值，然后调用NioEventLoop#run方法（该方法之后会详细分析）

// SingleThreadEventExecutor#doStartThread方法节选
executor.execute(new Runnable() {
    @Override
    public void run() {
        thread = Thread.currentThread();
        if (interrupted) {
            thread.interrupt();
        }

        boolean success = false;
        updateLastExecutionTime();
        // 调用NioEventLoop#run方法
        SingleThreadEventExecutor.this.run();
    }
}
// ThreadPerTaskExecutor#execute方法创建线程并运行
// ThreadPerTaskExecutor#execute方法
@Override
public void execute(Runnable command) {
    threadFactory.newThread(command).start();
}

// 在提交任务之后会唤醒阻塞的selector
// SingleThreadEventExecutor#startThread方法节选
wakeup(inEventLoop);
// NioEventLoop#wakeup方法
protected void wakeup(boolean inEventLoop) {
    if (!inEventLoop && wakenUp.compareAndSet(false, true)) {
        selector.wakeup();
    }
}
// 至此，`NioEventLoop`持有的线程就算创建完毕，也就是`Netty`中`bossGroup`的`NIO`线程已经创建完毕，那么此时就是执行刚才我们异步提交的任务的时候，比如提交的`Accptor`和`channel`注册

// 在线程创建完毕后回到AbstractChannel#register0方法，此时进行channel的注册，值得注意的时，此时的注册已经不是由主线程操作，而是NIO线程进行注册
// AbstractChannel#register0方法节选
// AbstractNioChannel#doRegister方法
doRegister();

pipeline.invokeHandlerAddedIfNeeded();

pipeline.fireChannelActive();

// 在AbstractChannel#register0中首先会调用AbstractNioChannel#doRegister方法，其中做的事是将channel注册到unwrappedSelector上
// AbstractNioChannel#doRegister方法节选
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

// 当把channel注册到selector上之后，会执行pipeline.invokeHandlerAddedIfNeeded()方法，该方法作用是执行之前我们所加的handler，根据之前的分析，目前在pipeline中还有一个Acceptor的异步方法没执行
pipeline.invokeHandlerAddedIfNeeded();
// 也就是此时会进入channel初始化中加入的ServerBootstrapAcceptor
p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});
// 当加入Accptor之后，一个NioServerSocketChannel的pipeline中相当于有3个元素：head、acceptor和tail
// 至此，channel的register基本算告一段落，相比于channel的创建和初始化，channel的注册调用链要复杂很多，期间还涉及到线程的切换，稍不注意就会跑飞

/**********************************doBind0源码分析****************************************/
// AbstractBootstrap#doBind方法节选
doBind0(regFuture, channel, localAddress, promise);
// 注意此时doBind0方法已经是由NIO线程来执行，AbstractBootstrap#doBind0方法会向NioEventLoop的任务队列中添加一个端口绑定任务
// AbstractBootstrap#doBind0方法节选
channel.eventLoop().execute(new Runnable() {
    @Override
    public void run() {
        if (regFuture.isSuccess()) {
            // NIO线程执行端口绑定工作
            channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        } else {
            promise.setFailure(regFuture.cause());
        }
    }
});
// 由于bind方法的调用链比较长，这里省略部分无关的调用，从DefaultChannelPipeline#bind方法开始分析，这里的unsafe是在创建channel的时候创建的，最后调用链来到AbstractUnsafe#bind方法
// DefaultChannelPipeline#bind方法
public void bind(
    ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
unsafe.bind(localAddress, promise);
}

// AbstractUnsafe#bind方法节选，NioServerSocketChannel#doBind方法内部也是调用的JDK的端口绑定方法
doBind(localAddress);
// NioServerSocketChannel#doBind方法节选
javaChannel().bind(localAddress, config.getBacklog());

// 可见在AbstractUnsafe#bind方法中完成绑定后会调用pipeline#fireChannelActive方法完成事件传播
// AbstractUnsafe#bind方法节选
if (!wasActive && isActive()) {
    invokeLater(new Runnable() {
        @Override
        public void run() {
            pipeline.fireChannelActive();
        }
    });
}



/**********************************NioEventLoop#run源码分析****************************************/

// 在上一小节channel的注册源码分析部分已经看出，当程序使用executor创建好线程后会开始调用NioEventLoop#run方法
// SingleThreadEventExecutor#doStartThread方法节选
executor.execute(new Runnable() {
    @Override
    public void run() {
        thread = Thread.currentThread();
        if (interrupted) {
            thread.interrupt();
        }

        boolean success = false;
        updateLastExecutionTime();
        // 调用NioEventLoop#run方法
        SingleThreadEventExecutor.this.run();
    }
}


// NioEventLoop#run方法节选，这里省略了一些我认为不重要的部分
@Override
protected void run() {
    for (;;) {
        try {
            try {
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;

                case SelectStrategy.BUSY_WAIT:
                    // fall-through to SELECT since the busy-wait is not supported with NIO

                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                    // fall through
                default:
                }
            }
        }
        final int ioRatio = this.ioRatio;
        if (ioRatio == 100) {
            try {
                processSelectedKeys();
            } finally {
                // Ensure we always run tasks.
                runAllTasks();
            }
        } else {
            final long ioStartTime = System.nanoTime();
            try {
                processSelectedKeys();
            } finally {
                // Ensure we always run tasks.
                final long ioTime = System.nanoTime() - ioStartTime;
                runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
            }
        }
    }
}
/**
 * 从NioEventLoop#run方法中可以看到，进来就是一死循环，然后调用DefaultSelectStrategy#calculateStrategy进行一个策略选择
 */
// NioEventLoop#run方法节选
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks()))
// DefaultSelectStrategy#calculateStrategy方法的逻辑也比较简单，其就是判断当前任务队列是否为空来进行判断，如果队列不为空，则执行从NioEventLoop#selectNow
// 否则进入执行NioEventLoop#select方法
// DefaultSelectStrategy#calculateStrategy方法
public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
    return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
}

// NioEventLoop#selectNow调用的是selector#selectNow，该方法是一个非阻塞的方法，作用是选择就绪的IO事件，底层调用的是一个本地的poll0方法
// 下面重点分析NioEventLoop#select方法，由于select方法比较长，这里将分片分析

// 首先进来就是获取当前的时间，并计算selectDeadLineNanos，这个参数与定时任务队列有关（定时任务队列是一个优先队列，在提交定时任务时创建），具体实现见下面的SingleThreadEventExecutor#delayNanos，
// 如果队列为空则延时1S，如果不为空则会在计算当前定时任务是否快要截至，如果是，则会在for循环中进行处理并跳出select方法
// NioEventLoop#select方法节选
Selector selector = this.selector;
try {
    int selectCnt = 0;
    long currentTimeNanos = System.nanoTime();
    long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
}
// SingleThreadEventExecutor#delayNanos
protected long delayNanos(long currentTimeNanos) {
    ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
    if (scheduledTask == null) {
        // 返回1S
        return SCHEDULE_PURGE_INTERVAL;
    }
    // 与定时任务有关
    return scheduledTask.delayNanos(currentTimeNanos);
}

// 当上面的时间计算完之后进入select的for循环，第一件事就是判断是否有定时任务将要截至
// 如果发现当前的定时任务队列中有任务的截止事件快到了(<=0.5ms)，就跳出循环；跳出之前如果发现目前为止还没有进行过select操作（if (selectCnt == 0)），那么就调用一次selectNow()，该方法会立即返回，不会阻塞
// NioEventLoop#select方法节选
for (;;) {
    long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
    if (timeoutMillis <= 0) {
        if (selectCnt == 0) {
            selector.selectNow();
            selectCnt = 1;
        }
        break;
    }
    // ...
}
// 之后如果没有定时任务或者定时任务还没到时间，那么在for循环中继续往下执行，判断在select循环过程中是否有任务加入到任务队列，如果有则执行selectNow并跳出循环
// NioEventLoop#select方法节选
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
    selector.selectNow();
    selectCnt = 1;
    break;
}
// 如果任务队列为空，则继续执行，调用阻塞的selector#select方法，阻塞时间在定时队列为空时为1S，不为空则根据最先截至的任务决定
// NioEventLoop#select方法节选
int selectedKeys = selector.select(timeoutMillis);
// 之后如果轮询到了事件，或者线程发生了中断或者发生了空轮询Bug则跳出循环，否则开始新一轮循环
// 在NioEventLoop#select方法中解决了NIO中著名的空轮询Bug，netty 会在每次进行 `selector.select(timeoutMillis)` 之前记录一下开始时间`currentTimeNanos`，
// 在select之后记录一下结束时间，判断select操作是否至少持续了`timeoutMillis`秒（这里将`time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos`
// 改成`time - currentTimeNanos >= TimeUnit.MILLISECONDS.toNanos(timeoutMillis)`或许更好理解一些）,
// 如果持续的时间大于等于timeoutMillis，说明就是一次有效的轮询，重置`selectCnt`标志，否则，表明该阻塞方法并没有阻塞这么长时间，可能触发了jdk的空轮询bug，当空轮询的次数超过一个阀值的时候，默认是512，就开始重建selector


// 接下来是执行processSelectedKeys，在开始处理之前会定义一个ioRatio
// `ioRatio`代表处理IO时间所占的时间比例，默认是50%，也就是一半的时间用来处理IO事件，一半的时间用来处理普通任务

final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
    // 如果ioRatio设置为100，则全部处理，让全部的普通任务运行完
    try {
        processSelectedKeys();
    } finally {
        // Ensure we always run tasks.
        runAllTasks();
    }
} else {
    final long ioStartTime = System.nanoTime();
    try {
        // 处理IO事件
        processSelectedKeys();
    } finally {
        // Ensure we always run tasks.
        // ioTime代表执行IO时间所花费的时间
        final long ioTime = System.nanoTime() - ioStartTime;
        // 假设ioRatio是80，上面处理IO时间花费了8秒，那么下面运行普通任务的时间计算出来就是2秒
        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
    }
}
// NioEventLoop#processSelectedKeys方法最后调用的是NioEventLoop#processSelectedKeysOptimized方法
// NioEventLoop#processSelectedKeys方法
private void processSelectedKeys() {
    if (selectedKeys != null) {
        processSelectedKeysOptimized();
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}
// 在NioEventLoop#processSelectedKeysOptimized方法中会对得到的selectionKeys集合进行遍历，然后获取到关联的channel，根据连接或者读事件进行处理
// NioEventLoop#processSelectedKeysOptimized方法
private void processSelectedKeysOptimized() {
    for (int i = 0; i < selectedKeys.size; ++i) {
        // 1、轮询selectedKeys，并获得关联的channel
        final SelectionKey k = selectedKeys.keys[i];
        selectedKeys.keys[i] = null;
		// 拿到关联的channel
        final Object a = k.attachment();
        // 2、处理channel
        if (a instanceof AbstractNioChannel) {
            processSelectedKey(k, (AbstractNioChannel) a);
        } else {
            @SuppressWarnings("unchecked")
            NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
            processSelectedKey(k, task);
        }
        // 3、判断是否该再来次轮询
        if (needsToSelectAgain) {
            selectedKeys.reset(i + 1);

            selectAgain();
            i = -1;
        }
    }
}

// 轮询到selectedKey和关联的channel之后再调用NioEventLoop#processSelectedKey，在该方法中会根据事件的类型（连接事件或者读事件）进行不同的处理，这部分会在之后进行详细分析
// NioEventLoop#processSelectedKey方法节选
int readyOps = k.readyOps();
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

// 在NioEventLoop#run方法的最后调用的是SingleThreadEventExecutor#runAllTasks()方法
// SingleThreadEventExecutor#runAllTasks方法
protected boolean runAllTasks(long timeoutNanos) {
    // 定时任务队列的任务转移，下面会做分析
    fetchFromScheduledTaskQueue();
    Runnable task = pollTask();
    if (task == null) {
        afterRunningAllTasks();
        return false;
    }
    // 计算任务的超时时间，与IO事件的执行事件和ioRatio有关
    final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
    long runTasks = 0;
    long lastExecutionTime;
    for (;;) {
        // 任务执行，直接调用任务的run方法，比较简单，不做分析
        safeExecute(task);

        runTasks ++;

        // Check timeout every 64 tasks because nanoTime() is relatively expensive.
        // XXX: Hard-coded value - will make it configurable if it is really a problem.
        // 每执行64个任务检查一下是否到达了超时事件，如果是，则退出本次任务的执行，否则继续执行任务
        if ((runTasks & 0x3F) == 0) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            if (lastExecutionTime >= deadline) {
                break;
            }
        }

        task = pollTask();
        // 如果任务为空则退出执行
        if (task == null) {
            lastExecutionTime = ScheduledFutureTask.nanoTime();
            break;
        }
    }
    // 最后执行下tailTask中的收尾任务，下面会做分析
    afterRunningAllTasks();
    this.lastExecutionTime = lastExecutionTime;
    return true;
}
// 通过SingleThreadEventExecutor#runAllTasks方法分析，在该方法中主要做了以下三件事：1、将定时任务队列与MPSC队列进行聚合；2、执行任务，每执行64个任务，判断是否到达超时时间，如果超时则退出，否则继续执行
// 3、当任务执行完毕，执行tailTask队列中的收尾工作（如果有收尾任务的话）
// SingleThreadEventExecutor#fetchFromScheduledTaskQueue方法
private boolean fetchFromScheduledTaskQueue() {
    // 如果定时任务队列为空的话直接返回
    if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
        return true;
    }
    long nanoTime = AbstractScheduledEventExecutor.nanoTime();
    for (;;) {
        // 取出定时任务队列中到达定时时间的任务，如果没有则返回空（定时任务队列是优先队列，底层是基于数组的二叉堆，可以类比PriorityQueue）
        Runnable scheduledTask = pollScheduledTask(nanoTime);
        if (scheduledTask == null) {
            return true;
        }
        // 如果在转移到MPSC队列的过程中出现失败，则重新将任务添加回定时任务队列
        if (!taskQueue.offer(scheduledTask)) {
            // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
            scheduledTaskQueue.add((ScheduledFutureTask<?>) scheduledTask);
            return false;
        }
    }
}
// 通过上面对SingleThreadEventExecutor#fetchFromScheduledTaskQueue方法的分析，可以看出该方法主要是将定时任务队列中到达时间的任务转移到MPSC中，Netty出于谨慎的原因，在转移失败时还会将定时任务重新放回定时任务队列
// 需要注意的是，在创建NioEventLoop的时候并没有创建定时任务队列，定时任务队列的创建是在第一次提交定时任务的时候创建的，创建过程比较简单，这里就不分析了，具体可以看AbstractScheduledEventExecutor#schedule方法

// 在SingleThreadEventExecutor#runAllTasks方法的最后会调用SingleThreadEventLoop#afterRunningAllTasks方法，该方法的主要目的是在每次运行任务完运行tailTask中的收尾工作，目前没有用到
// NioEventLoop可以通过父类SingleTheadEventLoop#executeAfterEventLoopIteration方法向tailTasks中添加收尾任务，比如，你想统计一下一次执行一次任务循环花了多长时间就可以调用此方法

// SingleThreadEventLoop#afterRunningAllTasks方法
protected void afterRunningAllTasks() {
    runAllTasksFrom(tailTasks);
}
// SingleThreadEventExecutor#runAllTasksFrom方法
protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
    Runnable task = pollTaskFrom(taskQueue);
    if (task == null) {
        return false;
    }
    for (;;) {
        safeExecute(task);
        task = pollTaskFrom(taskQueue);
        if (task == null) {
            return true;
        }
    }
}

// 至此，SingleThreadEventExecutor#runAllTasks方法基本算是全部分析完毕了，这个方法总的来说比前两个方法简单很多，其主要做的任务也就之前说的三点：1、将定时任务队列与MPSC队列进行聚合；2、执行任务，每执行64个任务，判断是否到达超时时间，如果超时则退出，否则继续执行
// 3、当任务执行完毕，执行tailTask队列中的收尾工作（如果有收尾任务的话）

/**********************************新连接接入过程源码分析****************************************/
// 由之前对NioEventLoop#processSelectedKey方法的分析可知，其新连接接入的主要执行代码体现在unsafe#read方法上，实质上，对于新连接的接入调用的是NioMessageUnsafe#read方法（NioMessageUnsafe
// 属于AbstractNioMessageChannel的内部类），在新连接接入时SelectionKey的ops的值是16
// NioEventLoop#processSelectedKey方法节选
int readyOps = k.readyOps();
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}
// NioMessageUnsafe#read方法节选（参考【闪电侠】）
private final List<Object> readBuf = new ArrayList<Object>();

public void read() {
    assert eventLoop().inEventLoop();
    final ChannelPipeline pipeline = pipeline();
    // 1、ByteBuf句柄
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    do {
        // 2、底层调用JDK的accept方法将channel包装成NioSocketChannel放入readBuf
        int localRead = doReadMessages(readBuf);
        if (localRead == 0) {
            break;
        }
        if (localRead < 0) {
            closed = true;
            break;
        }
    } while (allocHandle.continueReading());
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        // 3、对readBuf中的channel进行处理
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    pipeline.fireChannelReadComplete();
}
// NioMessageUnsafe#read方法节选，内部调用的是config()实现
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
recvHandle = config().getRecvByteBufAllocator().newHandle();
// 是否还记得config，这个属于channel得配置信息在channel创建得时候被创建，前面说过，其作用与直接buf相关，Netty中得ByteBuf提供了池化的思想，可以实现ByteBuf的重用。这里通过config分配一个可自动增长或减少的缓冲区的handle

// NioMessageUnsafe#read方法节选，实质调用的是NioServerSocketChannel#doReadMessages
int localRead = doReadMessages(readBuf);
// NioServerSocketChannel#doReadMessages
protected int doReadMessages(List<Object> buf) throws Exception {
    // 底层是serverSocketChannel.accept();还是利用JDK的NIO来获取channel
    SocketChannel ch = SocketUtils.accept(javaChannel());
    // 将获取到的channel进行包装并设置非阻塞，添加到readBuf中，以便在read方法中进行处理
    buf.add(new NioSocketChannel(this, ch));
    return 1;
}

// NioMessageUnsafe#read方法节选，目的是对获得的NioSocketChannel进行处理，目前对于Server的流水线由之前分析可知，其主要的一个组件是ServerBootstrapAcceptor
pipeline.fireChannelRead(readBuf.get(i));
// 所以，接下来重点关注ServerBootstrapAcceptor#channelRead方法
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;

    child.pipeline().addLast(childHandler);
    // 对channel进行一些属性设置
    setChannelOptions(child, childOptions, logger);
    setAttributes(child, childAttrs);

    try {
        // 将channel注册到workGroup中的一个NioEventLoop持有的Selector上，这里的注册可以参考前面分析的注册，流程和思想基本是一样的，这里就不做分析
        // 不过需要注意的是，注册操作是异步执行的，也就是是由workerGroup中的线程来执行，根据之前注册操作的分析，在register0方法注册完之后会调用pipeline.invokeHandlerAddedIfNeeded()执行初始化操作
        // 对于workGroup的初始化操作实质就是执行在创建启动类时添加到childHandler中的初始化方法，最后调用pipeline#fireChannelActive关注读事件
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}
// AbstractChannel#register0方法
private void register0(ChannelPromise promise) {
    // 注册
    doRegister();
    // 这里相当于调用的是在创建启动类时添加的到childHandler中的初始化方法
    pipeline.invokeHandlerAddedIfNeeded();
    safeSetSuccess(promise);
    pipeline.fireChannelRegistered();
    // 最后调用fireChannelActive开始事件传播，并关注读事件
    pipeline.fireChannelActive();
}

/**********************************读数据源码分析****************************************/
// 当一个新连接被接入后会创建一个客户端channel并注册到workerGroup中的一个NioEventLoop上，此后，该线程会进入run函数的死循环，对于数据的读流程，也是从`NioEventLoop#processSelectedKey`中开始
// 出现区别（之前已经对run方法进行了详细分析）
// 当是读信息时，readOps的值是1（处理连接时是16），这里的unsafe#read的实际的调用是NioByteUnsafe#read
int readyOps = k.readyOps();
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

// NioByteUnsafe#read方法
public final void read() {
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    // 分配ByteBuf的缓冲区
    final ByteBufAllocator allocator = config.getAllocator();
    // 动态调整Bytebuf，并使用直接内存，而不使用堆内存
    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
	
    // 分配bytebuf
    byteBuf = allocHandle.allocate(allocator);
    // 将内容读取到bytebuf
    allocHandle.lastBytesRead(doReadBytes(byteBuf));

    // 最后调用socketChannel的pipeline进行读事件传播，比如执行在启动类中配置的childHandler中的channelRead方法
     pipeline.fireChannelRead(byteBuf);
}

/**********************************ByteBuf补充****************************************/
// 
